package hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;


import java.io.IOException;
import java.util.List;
import java.util.Map;

public class HBaseHomework {
    Configuration conf = null;
    Connection connection = null;
    HBaseAdmin admin = null;
    @Before
    public void init() throws IOException {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","linux100,linux101,linux102");
        conf.set("hbase.zookeeper.property.clientPort","2181");
        connection = ConnectionFactory.createConnection(conf);
    }

    @Test
    public void createTable() throws IOException {
        admin = (HBaseAdmin) connection.getAdmin();
        HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf("relations"));
        hTableDescriptor.addFamily(new HColumnDescriptor("friends"));
        admin.createTable(hTableDescriptor);
    }

    @Test
    public void addFriend() throws IOException {
        // uid1添加好友uid2
        HTable table = (HTable) connection.getTable(TableName.valueOf("relations"));
        Put put = new Put(Bytes.toBytes("uid1"));
        put.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid2"), Bytes.toBytes("uid2"));
        table.put(put);

        // 同时为uid2添加好友uid1
        Map.Entry<byte[], List<Cell>> entry = put.getFamilyCellMap().firstEntry();
        Put p = new Put(CellUtil.cloneValue(entry.getValue().get(0)));
        p.addColumn(CellUtil.cloneFamily(entry.getValue().get(0)), put.getRow(), put.getRow());
        table.put(p);

        table.close();
    }

    @Test
    public void deleteFriend() throws IOException {
        //删除uid1好友uid2
        HTable table = (HTable) connection.getTable(TableName.valueOf("relations"));
        Delete delete = new Delete(Bytes.toBytes("uid1"));
        delete.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid2"));
        table.delete(delete);

        //同时删除uid2的好友uid1
        Cell cell = delete.getFamilyCellMap().firstEntry().getValue().get(0);
        Delete d = new Delete(CellUtil.cloneQualifier(cell));
        d.addColumn(CellUtil.cloneFamily(cell), delete.getRow());
        table.delete(d);

        table.close();
    }

    @Test
    public void getTable() throws IOException {
        HTable table = (HTable) connection.getTable(TableName.valueOf("relations"));
        Get get = new Get(Bytes.toBytes("uid"));
        get.addFamily(Bytes.toBytes("friends"));
        Result result = table.get(get);
        Cell[] cells = result.rawCells();
        System.out.print(Bytes.toString(CellUtil.cloneRow(cells[0])) + ":");
        for (Cell cell : cells){
            System.out.print(Bytes.toString(CellUtil.cloneValue(cell)) + " ");
        }
        System.out.println();
        table.close();
    }

    @Test
    public void deleteTable() throws IOException {
        admin = (HBaseAdmin) connection.getAdmin();
        admin.disableTable(TableName.valueOf("relations"));
        admin.deleteTable(TableName.valueOf("relations"));
    }

    @After
    public void destroy(){
        if(admin!=null){
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        if(connection !=null){
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }
}
